package com.fuyun.rpc.client.common;

import com.fuyun.rpc.client.config.RpcApplication;
import com.fuyun.rpc.common.domain.RpcResponse;
import com.fuyun.rpc.common.domain.ServiceMetaInfo;
import com.fuyun.rpc.common.serializer.Serializer;
import com.fuyun.rpc.common.serializer.SerializerFactory;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @author: zrq1351
 * @date: 2025/2/20
 */
@Slf4j
public class RpcTcpService {

	private static final Vertx VERTX = Vertx.vertx();
	private static final NetClient NET_CLIENT = VERTX.createNetClient();

	/**
	 * 获取RPC响应
	 *
	 * @param node           服务元信息，包含服务的主机和端口等信息
	 * @param bytes          请求的字节数组
	 * @param rpcApplication RPC应用实例，用于获取配置信息
	 * @return 返回RPC响应对象，如果请求失败或解析错误则返回null
	 */
	public static RpcResponse getRpcResponse(ServiceMetaInfo node, byte[] bytes, RpcApplication rpcApplication) {
		if (!isValidServiceMetaInfo(node)) {
			return null;
		}
		long timeout =
			rpcApplication.getRpcConfig().getTimeout() > 0 ? rpcApplication.getRpcConfig().getTimeout() : 10000;
		String serializerType = rpcApplication.getRpcConfig().getSerializer();
		Serializer serializer = SerializerFactory.getInstance(serializerType);

		if (serializer == null) {
			log.error("Serializer not found for type: {}", serializerType);
			return null;
		}

		CompletableFuture<RpcResponse> future = new CompletableFuture<>();

		// 设置连接超时
		NET_CLIENT.connect(node.getServicePort(), node.getServiceHost(), ar -> {
			if (ar.succeeded()) {
				NetSocket socket = ar.result();
				handleSocket(socket, bytes, serializer, future);
			} else {
				handleConnectionFailure(ar.cause(), future);
			}
		});

		try {
			return future.orTimeout(timeout, TimeUnit.MILLISECONDS).exceptionally(ex -> handleRpcException(ex)).join();
		} catch (Exception e) {
			log.error("RPC request failed: {}", e.getMessage(), e);
			return null;
		}
	}

	/**
	 * 检查服务元信息是否有效 有效性检查包括： 1. 服务元信息对象不能为空 2. 服务主机名不能为空 3. 服务端口号必须大于0
	 *
	 * @param node 服务元信息对象
	 * @return 如果服务元信息有效返回true，否则返回false
	 */
	private static boolean isValidServiceMetaInfo(ServiceMetaInfo node) {
		if (node == null || node.getServiceHost() == null || node.getServicePort() <= 0) {
			log.error("Invalid service meta info: {}", node);
			return false;
		}
		return true;
	}

	/**
	 * 处理Socket连接 该方法负责通过Socket发送请求，并设置响应、异常和关闭连接的处理程序
	 *
	 * @param socket       Socket连接对象，用于网络通信
	 * @param requestBytes 请求数据的字节表示，准备发送给服务端
	 * @param serializer   序列化器，用于响应数据的反序列化
	 * @param future       用于接收异步调用的结果，这里用于接收RPC响应
	 */
	private static void handleSocket(NetSocket socket, byte[] requestBytes, Serializer serializer,
		CompletableFuture<RpcResponse> future) {
		// 确保操作在正确的线程中执行
		VERTX.runOnContext(v -> {
			socket.write(Buffer.buffer(requestBytes));
			socket.handler(buffer -> processResponse(buffer, serializer, future, socket));
			socket.exceptionHandler(ex -> handleSocketException(ex, future, socket));
			socket.closeHandler(v2 -> log.debug("TCP connection closed"));
		});
	}

	/**
	 * 处理接收到的响应数据 本方法负责将接收到的缓冲区数据反序列化为RpcResponse对象，并通过CompletableFuture来通知等待结果的调用方
	 * 同时，它也处理反序列化过程中的异常情况，并确保在处理完成后关闭网络套接字
	 *
	 * @param buffer     缓冲区对象，包含接收到的数据
	 * @param serializer 序列化器，用于反序列化数据
	 * @param future     用于通知调用方处理结果的CompletableFuture对象
	 * @param socket     网络套接字，用于与远程节点通信
	 */
	private static void processResponse(Buffer buffer, Serializer serializer, CompletableFuture<RpcResponse> future,
		NetSocket socket) {
		try {
			RpcResponse response = serializer.deserialize(buffer.getBytes(), RpcResponse.class);
			log.info("Received remote service response: {}", response);
			future.complete(response);
		} catch (Exception e) {
			log.error("Deserialization error: {}", e.getMessage(), e);
			future.completeExceptionally(e);
		} finally {
			closeSocket(socket);
		}
	}

	private static void handleConnectionFailure(Throwable throwable, CompletableFuture<RpcResponse> future) {
		log.error("TCP connection failed: {}", throwable.getMessage(), throwable);
		future.completeExceptionally(throwable);
	}

	/**
	 * 处理Socket异常 当发生Socket异常时，此方法会被调用，它负责记录错误日志，将异常通知给未来的处理，并关闭Socket连接
	 *
	 * @param ex     异常对象，用于记录错误详情和完成未来对象的异常处理
	 * @param future 用于异步处理的CompletableFuture对象，通过它来通知调用者异常情况
	 * @param socket 发生异常的NetSocket对象，需要被关闭以释放资源
	 */
	private static void handleSocketException(Throwable ex, CompletableFuture<RpcResponse> future, NetSocket socket) {
		log.error("Socket error: {}", ex.getMessage(), ex);
		future.completeExceptionally(ex);
		closeSocket(socket);
	}

	/**
	 * 处理RPC异常 当RPC操作失败时调用此方法记录错误信息并返回null
	 *
	 * @param ex 异常对象，用于记录错误的详细信息
	 * @return 由于异常，RPC操作失败，返回null表示无响应
	 */
	private static RpcResponse handleRpcException(Throwable ex) {
		log.error("RPC operation failed: {}", ex.getMessage(), ex);
		return null;
	}

	/**
	 * 安全地关闭指定的 NetSocket。 该方法旨在关闭 Vert.x 上下文中的套接字，以确保线程安全和适当的资源释放。 它在套接字关闭过程中处理潜在的异常，并记录相应的警告。
	 *
	 * @param socket 要关闭的 NetSocket 实例。 这是需要释放的 socket 资源，可能为 null。
	 */
	private static void closeSocket(NetSocket socket) {
		VERTX.runOnContext(v -> {
			try {
				if (socket != null) {
					socket.close();
					log.debug("Socket closed successfully");
				}
			} catch (Exception e) {
				log.warn("Error closing socket: {}", e.getMessage());
			}
		});
	}
}